package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp;
import com.atguigu.realtime.bean.TradeProvinceOrderWindow;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.function.DimAsyncFunction;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2023/3/21 15:21
 */
public class Dws_10_DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_10_DwsTradeProvinceOrderWindow().init(
            4010,
            2,
            "Dws_10_DwsTradeProvinceOrderWindow",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
        
    }
    @Override
    public void handle(StreamExecutionEnvironment env, 
                       DataStreamSource<String> stream) {
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream = parseToPojo(stream);
    
        beanStream = distinctByOrderDetailId(beanStream);
    
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanWithoutDims = windowAndAgg(beanStream);
    
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanWithDims = joinDim(beanWithoutDims);
    
        writeToClickHouse(beanWithDims);
    
    
    }
    
    private void writeToClickHouse(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanWithoutDims) {
        beanWithoutDims.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> joinDim(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanWithoutDims) {
       return AsyncDataStream.unorderedWait(
            beanWithoutDims,
            new DimAsyncFunction<TradeProvinceOrderWindow>() {
                @Override
                public void addDim(TradeProvinceOrderWindow bean, JSONObject dim) {
                    bean.setProvinceName(dim.getString("NAME"));
                }
    
                @Override
                public String getId(TradeProvinceOrderWindow bean) {
                    return bean.getProvinceId();
                }
                
                @Override
                public String getTable() {
                    return "dim_base_province";
                }
            },
            60,
            TimeUnit.SECONDS
        );
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAndAgg(
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(TradeProvinceOrderWindow::getProvinceId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1,
                                                    TradeProvinceOrderWindow value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void process(String skuId,
                                        Context ctx,
                                        Iterable<TradeProvinceOrderWindow> elements,
                                        Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow bean = elements.iterator().next();
                        
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                        
                        bean.setTs(System.currentTimeMillis());
                        
                        bean.setOrderCount((long) bean.getOrderIdSet().size());
                        
                        
                        out.collect(bean);
                    }
                }
            );
    }
    
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> distinctByOrderDetailId(
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
        return beanStream
            .keyBy(TradeProvinceOrderWindow::getOrderDetailId)
            .process(new KeyedProcessFunction<String, TradeProvinceOrderWindow, TradeProvinceOrderWindow>() {
                
                private ValueState<TradeProvinceOrderWindow> state;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<TradeProvinceOrderWindow> desc =
                        new ValueStateDescriptor<>("beanState", TradeProvinceOrderWindow.class);
                    StateTtlConfig ttlConfig = new StateTtlConfig.Builder(org.apache.flink.api.common.time.Time.seconds(10))
                        .updateTtlOnCreateAndWrite()
                        .build();
                    
                    desc.enableTimeToLive(ttlConfig);
                    state = getRuntimeContext().getState(desc);
                }
                
                @Override
                public void processElement(TradeProvinceOrderWindow currentBean,
                                           Context ctx,
                                           Collector<TradeProvinceOrderWindow> out) throws Exception {
                    TradeProvinceOrderWindow lastBean = state.value();
                    if (lastBean == null) { // 当前详情 id 第一条数据
                        out.collect(currentBean);
                        //                        state.update(currentBean);
                    } else { // 不是第一条
                        // 用新的数据, 减去状态中的数据, 输出
                        lastBean.setOrderAmount(currentBean.getOrderAmount().subtract(lastBean.getOrderAmount()));
                        
                        out.collect(lastBean);
                        // 把新的数据存入到状态中
                        //                        state.update(currentBean);
                    }
                    state.update(currentBean);
                }
            });
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> parseToPojo(DataStreamSource<String> stream) {
       return stream.map(new MapFunction<String, TradeProvinceOrderWindow>() {
            @Override
            public TradeProvinceOrderWindow map(String value) throws Exception {
                JSONObject obj = JSON.parseObject(value);
                
//                /HashSet<String> orderIdSet = new HashSet<>();
//                orderIdSet.add(obj.getString("order_id"));
    
                return TradeProvinceOrderWindow.builder()
                    .orderDetailId(obj.getString("id"))
                    .provinceId(obj.getString("province_id"))
                    .orderIdSet(new HashSet<>(Collections.singletonList(obj.getString("order_id"))))
                    .orderAmount(obj.getBigDecimal("split_total_amount"))
                    .ts(obj.getLong("ts") * 1000)
                    .build();
            }
            
        });
    }
}
/*
如何统计下单数:
  1.  按照 order_id 分个组: 使用状态
    order_detail_id    order_id   金额   订单数
       1                10        100    1  第一个设置为 1
       1                10         0     0  其他设置为 0
       2                10        200    0
       3                10        300    0
   
   2.  只要能统计出有多个不同的 订单id , 把订单 id 去重, 然后计数
         把订单 id 存入到 set 集合中, 最后统计结果, 查看下 set 的长度
        
   
 */